package com.wenx.jdbccamel;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.camel.component.ActiveMQComponent;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.impl.SimpleRegistry;
import org.apache.camel.model.ModelCamelContext;
import org.apache.commons.dbcp.BasicDataSource;

import javax.jms.ConnectionFactory;

/**
 * jdbc查询
 */
public class PGDemo {
    private static String user = "admin";
    private static String password = "admin";
    //填写AMQ的IP
    private static String url = "tcp://10.10.3.186:61616";
    public static void main(String[] args) throws Exception {
        final String url = "jdbc:postgresql://10.15.0.208:5432/dmp";
        BasicDataSource basicDataSource = new BasicDataSource();
        basicDataSource.setDriverClassName("org.postgresql.Driver");
        basicDataSource.setUsername("dev");
        basicDataSource.setPassword("1");
        basicDataSource.setUrl(url);
        SimpleRegistry simpleregistry = new SimpleRegistry();
        simpleregistry.put("DataSource", basicDataSource);

        ModelCamelContext camelContext = new DefaultCamelContext(simpleregistry);
        //创建AMQ的连接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
        ActiveMQComponent activeMQComponent = new ActiveMQComponent();
        activeMQComponent.setConnectionFactory(connectionFactory);
        //使用连接池连接
        activeMQComponent.setUsePooledConnection(true);
        //事务
        activeMQComponent.setTransacted(true);
        //客户端确认
        activeMQComponent.setAcknowledgementMode(3);
        //持久化
        activeMQComponent.setDeliveryPersistent(true);
        camelContext.addComponent("activemq", activeMQComponent);
        camelContext.start();
        camelContext.addRoutes(new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("timer://queryAward?period=60s")
                        .doTry()
                        .setBody(constant("select * from common_menu"))
                        .to("jdbc:DataSource?outputType=SelectList")
                        .process(new Processor() {
                            @Override
                            public void process(Exchange exchange) throws Exception {
                                String str = exchange.getIn().getBody().toString();
                                System.out.println("str = " + str);
                            }
                        })
                        .to("direct:B")
                        .doCatch(Exception.class)
                        .process(new Processor() {
                            @Override
                            public void process(Exchange exchange) throws Exception {
                                System.out.println("异常");
                            }
                        });
                from("direct:B")
                        .to("activemq:topic:abcd");
            }
        });
        synchronized (PGDemo.class) {
            PGDemo.class.wait();
        }
    }
}
